#![recursion_limit="128"]
#![feature(slice_patterns)]

#[macro_use] extern crate serde_derive;
extern crate serde;
extern crate serde_json;
extern crate bincode;

extern crate chrono;

//extern crate itertools;

extern crate hyper;
extern crate hyper_tls;
extern crate futures;
extern crate tokio;

extern crate nanomsg;
extern crate either;

//#[path="../../src/models.rs"]
mod models;
pub use models::{Bucket, BucketRequest};

mod download;

mod disk;
pub use disk::{
  read_buckets,
  write_buckets
};

pub const BUCKET_REQ_ADDR: &str = "ipc:///pnf/bucket_req.ipc";

use nanomsg::{Socket, Protocol, Error};
use std::{thread, time};
use std::io::{Write};

use bincode::{serialize, deserialize};

use tokio::runtime::*;

fn main() {

  let mut socket = Socket::new(Protocol::Rep).unwrap();
  let _rec_endpoint = socket.bind(BUCKET_REQ_ADDR);
  let _resp_endpoint = socket.connect(BUCKET_REQ_ADDR);

  let mut buffer = Vec::new();
  let mut runtime = current_thread::Runtime::new().unwrap();
  let mut is_waiting = false;
  loop {
    match socket.nb_read_to_end(&mut buffer) {
      Ok(_) => {
        is_waiting = false;
        //println!("Read message of {} bytes!", buffer.len());
        let buckets = {
          let bucket_req: BucketRequest = deserialize(&buffer[..]).unwrap();
          buffer.clear();
          println!("Got: {:?}", bucket_req);
          let fut = get_n_last(bucket_req);
          runtime.block_on(Box::new(fut))
            .unwrap_or_else(
              |_err| panic!("bucket retrieval failed")
              )
          /*
          let bs_in_disk = disk::read_buckets(&bucket_req);
          bs_in_disk
              */
        };
        let encoded: Vec<u8> = serialize(&buckets).unwrap();
        match socket.write(&encoded) {
          Ok(..) => println!("Message sent!"),
          Err(err) => panic!("Failed to write to the socket: {}", err)
        }
      },
      Err(Error::TryAgain) => {
        //println!("Nothing to be read for the moment ...");
        if !is_waiting {
        println!("Waiting for a bucket request...");
        is_waiting = true;
        }
      },
      Err(err) => panic!("Unexpected error while reading socket: {}", err)
    };
    thread::sleep(time::Duration::from_millis(800));
  }
}

fn get_newest(bs: &Vec<Bucket>) -> Option<&Bucket> {
  bs.iter().max()
}

fn calc_n_latest_buckets_to_download(newest_persisted: &Bucket) -> u32 {
  use chrono::{Duration,Utc};
  let period_in_minutes = 5;
  let gap: Duration = Utc::now() - newest_persisted.timestamp;
  let gap_in_minutes_floored = gap.num_minutes() / period_in_minutes;
  // - better to fetch too much than too little
  // - first additional bucket is in case a new bucket has formed
  // since (Utc.now() - newest_persisted)
  // - second additional bucket is in case a new bucket has formed
  // since Utc.now()
  let compensated = gap_in_minutes_floored + 1 + 1;
  assert!(compensated > 0);
  compensated as u32
}

use futures::Future;

fn get_n_last(br: BucketRequest) -> impl Future<Item=Vec<Bucket>,Error=()> {
  let bs_in_disk = disk::read_buckets(&br);
  let requested_count = br.count;
  let n_latest_buckets_to_download = {
    let newest_in_disk = get_newest(&bs_in_disk);
    if let Some(newest_in_disk) = newest_in_disk {
      calc_n_latest_buckets_to_download(newest_in_disk)
    } else {
      requested_count
    }
  };
  println!("n_latest_buckets_to_download {}", n_latest_buckets_to_download);
  let br_clone = br.clone();
  download::get_n_last(&br.symbol, n_latest_buckets_to_download)
    .map(
      | dled_bs |
      {
        use std::collections::HashSet;
        fn to_set(bs: Vec<Bucket>) -> HashSet<Bucket> {
          bs.into_iter().collect()
        }
        let mut union: Vec<Bucket> = {
          let a = bs_in_disk;
          let b = dled_bs;
          let (a, b) = (to_set(a), to_set(b));
          a.union(&b).cloned().collect()
        };
        //let mut union: Vec<Bucket> = union(bs_in_disk, dled_bs).collect();
        union.sort_unstable();
        println!("total n buckets got {}", union.len());
        union
      }
      )
    .inspect(
      move | sorted_buckets |
      {
        disk::write_buckets(&br_clone, sorted_buckets)
      }
      )
}
